#include "MQTTClient.h"

MQTTClient::MQTTClient(const string &pubTopic, const string &subTopic, int qos)
    : pubTopic(pubTopic),
      subTopic(subTopic),
      qos(qos)
{
	int ret = mosquitto_lib_init();
	if(ret != MOSQ_ERR_SUCCESS)
	{
		fprintf(stderr, "Fail to mosquitto_lib_init,errmsg:%s\n", mosquitto_strerror(ret));
		return;
	}

	mos = mosquitto_new(NULL, true, this);
	if(!mos)
	{
		fprintf(stderr, "Fail to mosquitto_new,errmsg:%s\n", mosquitto_strerror(ret));
		return;
	}

	//回调函数
	mosquitto_message_callback_set(mos, message);
}

MQTTClient::~MQTTClient()
{
	mosquitto_destroy(mos);
	int ret = mosquitto_lib_cleanup();
	if(ret != MOSQ_ERR_SUCCESS)
	{
		fprintf(stderr, "Fail to mosquitto_lib_cleanup,errmsg:%s\n", mosquitto_strerror(ret));
	}
}

bool MQTTClient::connect(const string &ip, uint16_t port)
{
	int ret = mosquitto_connect(mos, ip.c_str(), port, 60);
	if(ret != MOSQ_ERR_SUCCESS)
	{
		fprintf(stderr, "Fail to mosquitto_connect,errmsg:%s\n", mosquitto_strerror(ret));
		return false;
	}

	//消息订阅
	ret = mosquitto_subscribe(mos, NULL, subTopic.c_str(), qos);
	if(ret != MOSQ_ERR_SUCCESS)
	{
		fprintf(stderr, "Fail to mosquitto_subscribe,errmsg:%s\n", mosquitto_strerror(ret));
		return false;
	}

	ret = mosquitto_loop_start(mos);
	if(ret != MOSQ_ERR_SUCCESS)
	{
		fprintf(stderr, "Fail to mosquitto_loop_start,errmsg:%s\n", mosquitto_strerror(ret));
		return false;
	}

	return true;
}

bool MQTTClient::disconnect(void)
{
	int ret = mosquitto_disconnect(mos);
	if(ret != MOSQ_ERR_SUCCESS)
	{
		fprintf(stderr, "Fail to mosquitto_disconnect,errmsg:%s\n", mosquitto_strerror(ret));
		return false;
	}

	ret = mosquitto_loop_stop(mos, true);
	if(ret != MOSQ_ERR_SUCCESS)
	{
		fprintf(stderr, "Fail to mosquitto_loop_stop,errmsg:%s\n", mosquitto_strerror(ret));
		return false;
	}

	return true;
}

int MQTTClient::send(const void *data, uint32_t len)
{
	int ret = mosquitto_publish(mos, NULL, pubTopic.c_str(), len, data, qos, false);
	if(ret != MOSQ_ERR_SUCCESS)
	{
		fprintf(stderr, "Fail to mosquitto_publish,errmsg:%s\n", mosquitto_strerror(ret));
		return -1;
	}

	return len;
}

int MQTTClient::recv(void *data, uint32_t len)
{
	qmumtx.lock();
	//没有消息则阻塞，有消息则获取消息
	if(msgQueue.empty())
	{
		qmumtx.unlock();
		wait();
		qmumtx.lock();
	}
	MQTTMessage msg = msgQueue.front();
	printf("msg topic:%s\n", msg.topic());
	printf("msg data:%s\n", (char *)msg.data());
	printf("msg size:%d\n", msg.size());
	printf("len:%d\n", len);
	//内存空间不够情况
	//获取消息的长度
	if(msg.size() > len)
	{
		qmumtx.unlock();
		return -1;
	}
	memcpy(data, msg.data(), msg.size());
	msgQueue.pop();
	qmumtx.unlock();

	return msg.size();
}

//收到消息的回调函数
void MQTTClient::message(mosquitto *, void *obj, const mosquitto_message *msg)
{
	cout << "========消息回调函数打印的================" << endl;
	printf("msg topic:%s\n", msg->topic);
	printf("msg data:%s\n", (char *)msg->payload);
	MQTTClient *client = (MQTTClient *)obj;

	client->qmumtx.lock();
	//插入队列
	client->msgQueue.push(MQTTMessage(msg));
	client->qmumtx.unlock();

	client->notify();
}

//等待
void MQTTClient::wait(void)
{
	unique_lock<mutex> lock(mtx);
	cv.wait(lock);
}

//唤醒
void MQTTClient::notify(void)
{
	cv.notify_one();
}
